package com.gitee.sparta.core.listener;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.gitee.sparta.core.bean.cluster.ClusterCmd;
import com.gitee.sparta.common.IdentityEnum;
import com.gitee.sparta.common.SpartaLuaConstant;
import com.gitee.sparta.common.helpers.CollHelper;
import com.gitee.sparta.common.helpers.IpHelper;
import com.gitee.sparta.common.helpers.IsolationHelper;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.bean.cluster.ServerBean;
import com.gitee.sparta.core.bean.event.LeaderEnableEvent;
import com.gitee.sparta.core.bean.event.SingleServerEnableEvent;
import com.gitee.sparta.core.bean.event.SpartaRefreshEvent;
import com.gitee.sparta.core.cluster.AbstractClusterFollower;
import com.gitee.sparta.core.cluster.AbstractClusterLeader;
import com.gitee.sparta.core.cluster.ClusterCache;
import com.gitee.sparta.core.cluster.balance.LoadBalance;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;

import javax.annotation.PostConstruct;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;

/**
 * <p>
 *      sparta 启动完毕事件的监听
 *      主要是针对 readable_tab 中上一次关机后，剩余的已经读取但是未被删除的数据进行处理
 *      重新调用回调用户添加了 @SpartaListener 方法，然后执行完整的逻辑进行处理完毕
 * </p>
 *
 * @author zyred
 * @since v 1.0.0
 */
@Slf4j
public class SpartaRefreshEventListener implements ApplicationListener<ContextRefreshedEvent>, ApplicationContextAware {

    private final LoadBalance loadBalance;
    private final RedissonClient client;
    private final SpartaCoreProperties properties;
    private final ApplicationEventPublisher eventPublisher;

    public SpartaRefreshEventListener(LoadBalance loadBalance,
                                      RedissonClient client,
                                      SpartaCoreProperties properties,
                                      ApplicationEventPublisher eventPublisher) {
        this.loadBalance = loadBalance;
        this.client = client;
        this.properties = properties;
        this.eventPublisher = eventPublisher;
    }

    private ApplicationContext applicationContext;
    private AbstractClusterLeader leader;
    private AbstractClusterFollower follower;

    @PostConstruct
    public void init() {
        // 如果开启了集群，才初始化 bean
        // 否则直接加载会报错的
        if (!this.properties.isStandalone()) {
            this.leader = this.applicationContext.getBean(AbstractClusterLeader.class);
            this.follower = this.applicationContext.getBean(AbstractClusterFollower.class);
        }
    }


    @SuppressWarnings("all")
    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        // 注册脚本摘要
        this.registerScriptSha(
                this.client.getMap(this.properties.getShaTable())
        );

        // 开启集群配置后，才会构建集群的通讯
        if (this.properties.isStandalone()) {
            // 服务注册
            this.registrationServer();
        } else {
            this.startStandaloneServer();
        }
        // 发布启动完毕事件
        this.eventPublisher.publishEvent(new SpartaRefreshEvent(properties));
    }

    private void startStandaloneServer() {
        // 单机服务启动，需要提前构造 load balance
        int sid = CollHelper.lackNum(Collections.emptySet());
        String isolation = IsolationHelper.isolation(sid);
        this.loadBalance.addNode(isolation, sid);
        ClusterCache.setLeaderSid(sid);
        ClusterCache.setServerId(sid);
        ClusterCache.setServerIdentity(IdentityEnum.LEADER);
        ClusterCache.leaderIsolation = isolation;

        this.properties.setIsolation(isolation);

        ServerBean leader = new ServerBean();
        leader.setServerPort(0);
        leader.setIsolation(isolation);
        leader.setServerId(sid);
        leader.setServerIp("0.0.0.0");
        leader.setLeader(true);

        ClusterCache.putServerInfo(sid, leader);
        // 发布单机服务启动事件
        this.eventPublisher.publishEvent(new SingleServerEnableEvent(properties));
    }


    /**
     * 服务注册
     * 1. 注册服务编号表，用于解决服务集群部署的隔离问题
     * 2. 注册 lua 脚本 sha 摘要的保存，服务集群部署时，会多次 loadScript 到 redis，做无效工作
     */
    private void registrationServer() {
        RLock lock = this.client.getLock(this.properties.getLock());
        try {
            lock.lock();
            // 注册服务编号
            this.registerServer(
                    this.client.getMap(this.properties.getServerTable())
            );

        } finally {
            lock.unlock();
        }
    }


    /**
     * 服务注册
     *
     * @param serverTab 服务信息表
     */
    private void registerServer(RMap<Integer, ServerBean> serverTab) {
        final String addr = IpHelper.getAddress();

        // 无服务列表
        if (CollUtil.isEmpty(serverTab)) {
            log.info("[Sparta] Not find leader, try build self leader.");
            this.selfLeader(addr, serverTab);
            log.info("[Sparta] Build self leader success.");
        }

        // 有服务列表
        else {
            Collection<ServerBean> beans = serverTab.values();
            ServerBean bean = beans.stream().filter(ServerBean::isLeader).findFirst().orElse(null);

            if (Objects.nonNull(bean)) {
                try {
                    log.info("[Sparta] Try connect leader, leader ip : {}, port: {}",
                            bean.getServerIp(), bean.getServerPort());
                    this.follower.buildFollower(bean.getServerIp(), bean.getServerPort());
                    ClusterProtocol pro = ClusterProtocol.defaultValue(ClusterCmd.STOP_LEADER_SENDER);
                    AbstractClusterLeader.sendQueue(pro);
                    log.info("[Sparta] Connected leader success, self is {}.", ClusterCache.getServerIdentity());
                } catch (Exception ex) {
                    log.warn("[Sparta] Try connect leader fail: {}.", ex.getMessage());
                    // 删除连接失败的服务列表
                    serverTab.remove(bean.getServerId());
                    // 重新把自己建立成 leader
                    log.info("[Sparta] Try build self leader....");
                    this.selfLeader(addr, serverTab);
                    log.info("[Sparta] Build self leader success, self is leader. ");
                }
            }

            else {
                // 有可能会出现假象
                // beans 中有 leader，但是 isLeader=false(人为修改的)
                // 这种假象要处理

                boolean ask = false;
                for (ServerBean sb : beans) {
                    if (!StrUtil.hasBlank(sb.getServerIp(), String.valueOf(sb.getServerPort()))) {
                        try {
                            log.info("[Sparta] Ask try connect leader, leader ip : {}, port: {}",
                                    sb.getServerIp(), sb.getServerPort());

                            this.follower.buildFollower(sb.getServerIp(), sb.getServerPort());
                            log.info("[Sparta] Ask connect leader success, self is {}.", ClusterCache.getServerIdentity());
                            ask = true;
                            ClusterProtocol protocol = ClusterProtocol
                                    .defaultValue(ClusterCmd.ASK_LEADER);
                            AbstractClusterFollower.sendQueue(protocol);
                        } catch (Exception e) {
                            log.warn("[Sparta] Ask try connect leader connect fail, exception :", e);
                            ask = false;
                        }
                    }
                }

                // 去询问
                if (!ask) {
                    log.info("[Sparta] Ask leader fail, try build self leader ...");
                    this.selfLeader(addr, serverTab);
                    log.info("[Sparta] Build self leader success ...");
                }
            }
        }

        // 启动完毕后，判断自己是不是 leader
        if (Objects.equals(ClusterCache.getServerIdentity(), IdentityEnum.LEADER)) {
            this.eventPublisher.publishEvent(new LeaderEnableEvent(IdentityEnum.LEADER));
        }
    }


    /**
     * 构建自己为 leader
     *
     * @param addr          ip
     * @param serverTab     服务信息
     */
    public void selfLeader (String addr, RMap<Integer, ServerBean> serverTab) {
        int sid = CollHelper.lackNum(Collections.emptySet());
        ClusterCache.setServerId(sid);

        // leader 启动通讯线程
        this.leader.buildLeader();

        String isolation = IsolationHelper.isolation(sid);
        this.properties.setIsolation(isolation);

        ServerBean leader = new ServerBean();
        leader.setServerPort(this.properties.getCluster().getClusterTransportPort());
        leader.setIsolation(isolation);
        leader.setServerId(sid);
        leader.setServerIp(addr);
        leader.setLeader(true);

        serverTab.put(sid, leader);
        ClusterCache.putServerInfo(sid, leader);
        ClusterCache.setServerIdentity(IdentityEnum.LEADER);

        this.loadBalance.addNode(isolation, sid);
        ClusterCache.leaderIsolation = isolation;
        log.info("[Sparta] Self leader addr: {}, myId: {}, isolation: {}", addr, sid, isolation);
    }

    /**
     * 注册 redis script sha 摘要
     *
     * @param serviceTab 服务信息表
     */
    private void registerScriptSha(RMap<String, String> serviceTab) {
        Map<String, String> scriptLoaded = SpartaLuaConstant.loadScript(this.client);
        serviceTab.putAll(scriptLoaded);
    }

    @SuppressWarnings("all")
    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        this.applicationContext = ctx;
    }
}
